package com.wmx.rocketmq.helloWorld;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消费者 - 接收消息
 *
 * @author wangMaoXiong
 * @version 1.0
 * @date 2020/11/2 15:54
 */
public class Consumer {
    public Consumer() {
    }

    public static void main(String[] args) throws Exception {

        /**
         * 1、声明并初始化 一个 consumer 消费者，ConsumerGroupName 需要由应用来保证唯一
         * 2、Consumer Group 消费者组名称，比如多个 Consumer 属于一个应用，订阅同样的消息，且消费逻辑一致，则应该将它们归为同一组
         * 3、一个应用中创建一个 Consumer，由应用来维护此对象，可以设置为全局对象或者单例
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketMQ-helloWorld-consumerGroup");

        /**
         * 1、指定 Producer 连接的 nameServer 服务器所在地址以及端口
         * 2、如果是分布式部署的多个，则用分号隔开，如：setNamesrvAddr("172.16.235.77:9876;172.16.235.78:9876");
         */
        consumer.setNamesrvAddr("192.168.116.128:9876");

        /**
         * 1、消费者订阅消息，如下所示订阅 TopicTest（主题） 下的所有 tag（标签） 类型的消息
         * 2、一个 consumer 对象可以订阅多个 topic
         * 3、订阅指定 topic下 tags 分别等于 TagA 或 TagB 或 TagC，则为：<code>consumer.subscribe("TopicTest", "TagA || TagB || TagC")</code>;
         * RocketMQ 定义了一些基本语法
         *     数值比较，比如：>，>=，<，<=，BETWEEN，=；
         *     字符比较，比如：=，<>，IN；
         *     IS NULL 或者 IS NOT NULL；
         *     逻辑符号 AND，OR，NOT；
         * 常量支持类型为：
         *     数值，比如：123，3.1415；
         *     字符，比如：'abc'，必须用单引号包裹起来；
         *     NULL，特殊的常量
         *     布尔值，TRUE 或 FALSE
         */
        consumer.subscribe("TopicTest", "*");

        /**
         * 设置消费者从何处开始消费，{@link ConsumeFromWhere} 可选值有三个
         * 1）CONSUME_FROM_LAST_OFFSET：(默认方式)消费者客户机从上次读到的位置继续开始。如果是第一次订阅的消费者客户端，则有两种情况：
         * A）如果用户组创建得太晚，而订阅的消息还未过期，则消费者将从最开始的位置进行读取；
         * B）如果订阅的消息已过期，则消费将从最新消息开始，这意味着已经过期的消息将被忽略。
         *
         * 2）CONSUME_FROM_FIRST_OFFSET：消费者客户端将从队列最开始的位置进行消费,即历史消息只要未过期的都消费一遍
         *
         * 3）CONSUME_FROM_TIMESTAMP：消费者客户端将从指定的时间戳开始消费，这意味着在指定时间戳之前生成的消息将被忽略.
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //注册消息监听器，如果有订阅的消息就会触发响应。注册回调实现类来处理从broker拉取回来的消息
        //MessageListenerConcurrently 是无序的、MessageListenerOrderly 是有序的
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext context) {
                try {
                    System.out.printf("【%s】接收到消息: %s %n", Thread.currentThread().getName(), messageExtList);
                    /**
                     * 标记该消息是否被成功消费
                     * CONSUME_SUCCESS 消费成功
                     * RECONSUME_LATER 消费失败，需要稍后重新消费
                     */
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    //消费失败,后期需要再重新消费
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        //Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br>
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}